//
// Created by wayne on 18-11-25.
//

#ifndef ENGINE_RACE_RDPCACHE_H
#define ENGINE_RACE_RDPCACHE_H

#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <tgmath.h>
#include <thread>
#include "KeyLogCache.h"
#include "KeyValueLogManager.h"

namespace polar_race {

    class RDPCache {
    public:

        RDPCache(int cacheCount, int cacheThreadNumber, size_t cacheSize) : cacheCount(cacheCount),
                                                                            cacheSize(cacheSize) {
            caches = new KeyLogCache[cacheCount];
            pthread_mutex_init(&mutex, nullptr);
            pthread_cond_init(&condition, nullptr);
        }

        ~RDPCache() {
            delete[] caches;
            pthread_mutex_destroy(&mutex);
            pthread_cond_destroy(&condition);
        }

        KeyLogCache *getKeyLogCacheToRead(int keyLogId) {
            int cacheIndex = keyLogId % cacheCount;
            auto cache = (caches + cacheIndex);
            pthread_mutex_lock(&(cache->mutex));
            if (cache->keyLogId == keyLogId) {
                cache->inUsed++;
                pthread_mutex_unlock(&(cache->mutex));
                return cache;
            }
            if (cache->inUsed > 0) {
                pthread_mutex_unlock(&(cache->mutex));
                return nullptr;
            }
            cache->keyLogId = keyLogId;
            cache->partToFill = 0;
            cache->completedPart = 0;
            cache->inUsed = 1;
            pthread_mutex_unlock(&(cache->mutex));
            return cache;
        }


        void rangeCache(KeyValueLog *keyValueLog) {
            KeyLogCache *cache = nullptr;
            while (true) {
                cache = getKeyLogCacheToRead(keyValueLog->id);
                if (cache != nullptr) {
                    break;
                }
            }
            while (true) {
                int partToRead = cache->getPartToRead();
                if (partToRead == -1) {
                    break;
                }
                auto offset = partToRead * PerPartSize;
                pread64(keyValueLog->valueFd, static_cast<char *>(cache->cache) + offset, PerPartSize,
                        offset + keyValueLog->initialOffset);
                cache->setPartCompleted();
            }
            cache->releaseReference();
        }

        void startAllKeyLogsCache(std::vector<KeyValueLogManager *> pKeyValueLogs, int threadId) {

            //affinity thread.
            cpu_set_t mask;
            CPU_ZERO(&mask);
            CPU_SET(threadId, &mask);
            int rc = pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);

            //see [C++11 threads, affinity and hyper-threading](https://eli.thegreenplace.net/2016/c11-threads-affinity-and-hyperthreading/)
            if (rc != 0) {
                fprintf(stderr, "Error calling pthread_setaffinity_np: %d\n ", rc);
            }

            for (int i = 0; i < 2; i++) {
                for (int n = 0; n < KeyManagerNum; n++) {
                    for (int m = 0; m < KeyLogPerManagerNum; m++)
                        rangeCache(pKeyValueLogs[n]->keyValueLogs[m]);
                }
            }
        }

        void startCaching(std::vector<KeyValueLogManager *> p, int size) {
            pthread_mutex_lock(&mutex);
            if (!caching) { //double check
                for (int i = 0; i < CacheThreadNumber; i++) {
                    std::thread t(&polar_race::RDPCache::startAllKeyLogsCache, this, p, i);
                    t.detach();
                }
                caching = true;
            }
            pthread_mutex_unlock(&mutex);
        }

        bool caching = false;

        int cacheCount;
        size_t cacheSize;
        KeyLogCache *caches;
        pthread_mutex_t mutex;
        pthread_cond_t condition;
    };
}

#endif //ENGINE_RACE_RDPCACHE_H
